#include "ExecutionPlan.h"
#include "PredicatePushdownOptimizer.h"
#include "AST.h"
#include <sstream>
#include <iostream>
#include <stdexcept>
#include <algorithm>
#include <unordered_map>

namespace MiniDB {

// ================================
// 表达式求值器
// ================================

class ExpressionEvaluator {
public:
    explicit ExpressionEvaluator(const std::unordered_map<std::string, std::variant<std::string, int, double>>& row)
        : row_(row) {}

    std::variant<std::string, int, double> evaluate(std::shared_ptr<Expression> expr) {
        if (auto binary_expr = std::dynamic_pointer_cast<BinaryExpression>(expr)) {
            return evaluateBinary(binary_expr);
        }
        else if (auto unary_expr = std::dynamic_pointer_cast<UnaryExpression>(expr)) {
            return evaluateUnary(unary_expr);
        }
        else if (auto column_ref = std::dynamic_pointer_cast<ColumnReference>(expr)) {
            return evaluateColumnReference(column_ref);
        }
        else if (auto literal = std::dynamic_pointer_cast<Literal>(expr)) {
            return literal->value;
        }
        else {
            throw std::runtime_error("Unsupported expression type in WHERE clause");
        }
    }

private:
    std::variant<std::string, int, double> evaluateBinary(std::shared_ptr<BinaryExpression> expr) {
        auto left_val = evaluate(expr->left);
        auto right_val = evaluate(expr->right);

        if (expr->operator_ == "=" || expr->operator_ == "==") {
            return compareEqual(left_val, right_val);
        }
        else if (expr->operator_ == "!=") {
            return !compareEqual(left_val, right_val);
        }
        else if (expr->operator_ == ">") {
            return compareGreater(left_val, right_val);
        }
        else if (expr->operator_ == "<") {
            return compareLess(left_val, right_val);
        }
        else if (expr->operator_ == ">=") {
            return compareGreater(left_val, right_val) || compareEqual(left_val, right_val);
        }
        else if (expr->operator_ == "<=") {
            return compareLess(left_val, right_val) || compareEqual(left_val, right_val);
        }
        else if (expr->operator_ == "AND") {
            bool left_bool = toBoolean(left_val);
            bool right_bool = toBoolean(right_val);
            return left_bool && right_bool;
        }
        else if (expr->operator_ == "OR") {
            bool left_bool = toBoolean(left_val);
            bool right_bool = toBoolean(right_val);
            return left_bool || right_bool;
        }
        else {
            throw std::runtime_error("Unsupported binary operator: " + expr->operator_);
        }
    }

    std::variant<std::string, int, double> evaluateUnary(std::shared_ptr<UnaryExpression> expr) {
        auto operand_val = evaluate(expr->operand);

        if (expr->operator_ == "NOT") {
            bool operand_bool = toBoolean(operand_val);
            return !operand_bool;
        }
        else {
            throw std::runtime_error("Unsupported unary operator: " + expr->operator_);
        }
    }

    std::variant<std::string, int, double> evaluateColumnReference(std::shared_ptr<ColumnReference> expr) {
        auto it = row_.find(expr->column_name);
        if (it == row_.end()) {
            throw std::runtime_error("Column not found: " + expr->column_name);
        }
        return it->second;
    }

    bool compareEqual(const std::variant<std::string, int, double>& left,
                      const std::variant<std::string, int, double>& right) {
        if (left.index() != right.index()) {
            return false;
        }

        if (std::holds_alternative<std::string>(left)) {
            return std::get<std::string>(left) == std::get<std::string>(right);
        }
        else if (std::holds_alternative<int>(left)) {
            return std::get<int>(left) == std::get<int>(right);
        }
        else if (std::holds_alternative<double>(left)) {
            return std::get<double>(left) == std::get<double>(right);
        }

        return false;
    }

    bool compareGreater(const std::variant<std::string, int, double>& left,
                       const std::variant<std::string, int, double>& right) {
        if (left.index() != right.index()) {
            return false;
        }

        if (std::holds_alternative<std::string>(left)) {
            return std::get<std::string>(left) > std::get<std::string>(right);
        }
        else if (std::holds_alternative<int>(left)) {
            return std::get<int>(left) > std::get<int>(right);
        }
        else if (std::holds_alternative<double>(left)) {
            return std::get<double>(left) > std::get<double>(right);
        }

        return false;
    }

    bool compareLess(const std::variant<std::string, int, double>& left,
                    const std::variant<std::string, int, double>& right) {
        if (left.index() != right.index()) {
            return false;
        }

        if (std::holds_alternative<std::string>(left)) {
            return std::get<std::string>(left) < std::get<std::string>(right);
        }
        else if (std::holds_alternative<int>(left)) {
            return std::get<int>(left) < std::get<int>(right);
        }
        else if (std::holds_alternative<double>(left)) {
            return std::get<double>(left) < std::get<double>(right);
        }

        return false;
    }

    bool toBoolean(const std::variant<std::string, int, double>& val) {
        if (std::holds_alternative<std::string>(val)) {
            return !std::get<std::string>(val).empty();
        }
        else if (std::holds_alternative<int>(val)) {
            return std::get<int>(val) != 0;
        }
        else if (std::holds_alternative<double>(val)) {
            return std::get<double>(val) != 0.0;
        }

        return false;
    }

    const std::unordered_map<std::string, std::variant<std::string, int, double>>& row_;
};

// ================================
// 执行计划算子实现
// ================================

// 顺序扫描算子
SeqScanOperator::SeqScanOperator(const std::string& table_name)
    : table_name_(table_name) {}

std::string SeqScanOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    return prefix + "SeqScan(" + table_name_ + ")";
}

// 过滤算子
FilterOperator::FilterOperator(std::shared_ptr<PlanOperator> child, std::shared_ptr<Expression> condition)
    : child_(std::move(child)), condition_(std::move(condition)) {}

std::string FilterOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    std::stringstream ss;
    ss << prefix << "Filter(";
    if (condition_) {
        ss << condition_->toString();
    }
    ss << ")\n";
    if (child_) {
        ss << child_->toString(indent + 2);
    }
    return ss.str();
}

// 投影算子
ProjectOperator::ProjectOperator(std::shared_ptr<PlanOperator> child,
                                const std::vector<std::shared_ptr<Expression>>& projections)
    : child_(std::move(child)), projections_(projections) {}

std::string ProjectOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    std::stringstream ss;
    ss << prefix << "Project(";
    for (size_t i = 0; i < projections_.size(); ++i) {
        if (i > 0) ss << ", ";
        ss << projections_[i]->toString();
    }
    ss << ")\n";
    if (child_) {
        ss << child_->toString(indent + 2);
    }
    return ss.str();
}

// 插入算子
InsertOperator::InsertOperator(const std::string& table_name,
                              const std::vector<std::vector<std::shared_ptr<Expression>>>& values)
    : table_name_(table_name), values_(values) {}

std::string InsertOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    return prefix + "Insert(" + table_name_ + ")";
}

// 更新算子
UpdateOperator::UpdateOperator(const std::string& table_name,
                              const std::vector<std::pair<std::string, std::shared_ptr<Expression>>>& assignments,
                              std::shared_ptr<Expression> condition)
    : table_name_(table_name), assignments_(assignments), condition_(std::move(condition)) {}

std::string UpdateOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    return prefix + "Update(" + table_name_ + ")";
}

// 删除算子
DeleteOperator::DeleteOperator(const std::string& table_name, std::shared_ptr<Expression> condition)
    : table_name_(table_name), condition_(std::move(condition)) {}

std::string DeleteOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    return prefix + "Delete(" + table_name_ + ")";
}

// 创建表算子
CreateTableOperator::CreateTableOperator(const std::string& table_name,
                                        const std::vector<std::shared_ptr<ASTNode>>& columns)
    : table_name_(table_name), columns_(columns) {}

std::string CreateTableOperator::toString(int indent) const {
    std::string prefix(indent, ' ');
    return prefix + "CreateTable(" + table_name_ + ")";
}

// ================================
// 执行计划实现
// ================================

std::string ExecutionPlan::toString() const {
    if (!root_) {
        return "Empty Plan";
    }
    return root_->toString();
}

void ExecutionPlan::print() const {
    std::cout << "=== Execution Plan ===" << std::endl;
    std::cout << toString() << std::endl;
}

// ================================
// 查询执行器
// ================================

class QueryExecutor {
public:
    QueryExecutor() = default;

    // 执行SELECT查询
    std::vector<std::unordered_map<std::string, std::variant<std::string, int, double>>>
    executeSelect(std::shared_ptr<SelectStatement> select_stmt) {

        // 简化的表数据（实际应该从存储系统读取）
        std::vector<std::unordered_map<std::string, std::variant<std::string, int, double>>> table_data;

        // 模拟表数据 - 这里应该从实际的存储系统读取
        if (auto from_clause = std::dynamic_pointer_cast<FromClause>(select_stmt->from_clause)) {
            table_data = getTableData(from_clause->table_name);
        }

        std::vector<std::unordered_map<std::string, std::variant<std::string, int, double>>> result;

        // 应用WHERE条件过滤
        for (const auto& row : table_data) {
            bool matches = true;

            if (select_stmt->where_clause) {
                ExpressionEvaluator evaluator(row);
                try {
                    auto condition_result = evaluator.evaluate(select_stmt->where_clause);
                    matches = toBoolean(condition_result);
                } catch (const std::exception& e) {
                    std::cerr << "WHERE condition evaluation error: " << e.what() << std::endl;
                    matches = false;
                }
            }

            if (matches) {
                // 应用投影
                std::unordered_map<std::string, std::variant<std::string, int, double>> projected_row;

                for (const auto& expr : select_stmt->select_list) {
                    if (auto column_ref = std::dynamic_pointer_cast<ColumnReference>(expr)) {
                        if (column_ref->column_name == "*") {
                            // 选择所有列
                            projected_row = row;
                            break;
                        } else {
                            // 选择特定列
                            auto it = row.find(column_ref->column_name);
                            if (it != row.end()) {
                                projected_row[column_ref->column_name] = it->second;
                            }
                        }
                    } else if (auto literal = std::dynamic_pointer_cast<Literal>(expr)) {
                        // 处理字面量
                        projected_row["literal"] = literal->value;
                    }
                }

                result.push_back(projected_row);
            }
        }

        return result;
    }

private:
    // 获取表数据的简化实现（实际应该从存储管理器读取）
    std::vector<std::unordered_map<std::string, std::variant<std::string, int, double>>>
    getTableData(const std::string& table_name) {

        // 这里是模拟数据，实际应该从文件系统或数据库存储中读取
        if (table_name == "users") {
            return {
                {{"id", 1}, {"name", std::string("Alice")}, {"age", 25}, {"email", std::string("alice@example.com")}},
                {{"id", 2}, {"name", std::string("Bob")}, {"age", 30}, {"email", std::string("bob@example.com")}},
                {{"id", 3}, {"name", std::string("Charlie")}, {"age", 20}, {"email", std::string("charlie@example.com")}}
            };
        }

        return {};
    }

    bool toBoolean(const std::variant<std::string, int, double>& val) {
        if (std::holds_alternative<std::string>(val)) {
            return !std::get<std::string>(val).empty();
        }
        else if (std::holds_alternative<int>(val)) {
            return std::get<int>(val) != 0;
        }
        else if (std::holds_alternative<double>(val)) {
            return std::get<double>(val) != 0.0;
        }
        return false;
    }
};

// 全局查询执行器实例
static QueryExecutor global_executor;

// ================================
// 执行计划生成器实现
// ================================

PlanGenerator::PlanGenerator(SymbolTable& symbol_table)
    : symbol_table_(symbol_table) {}

std::unique_ptr<ExecutionPlan> PlanGenerator::generatePlan(const std::vector<std::shared_ptr<Statement>>& statements) {
    auto plan = std::make_unique<ExecutionPlan>();

    if (statements.empty()) {
        return plan;
    }

    // 生成初始执行计划
    auto root_operator = generateStatementPlan(statements[0]);
    plan->setRoot(root_operator);

    // 如果启用了谓词下推优化，则应用优化
    if (predicate_pushdown_enabled_) {
        PredicatePushdownOptimizer optimizer(symbol_table_);
        plan = optimizer.optimize(std::move(plan));
    }

    return plan;
}

std::shared_ptr<PlanOperator> PlanGenerator::generateStatementPlan(std::shared_ptr<Statement> stmt) {
    if (auto select_stmt = std::dynamic_pointer_cast<SelectStatement>(stmt)) {
        return generateSelectPlan(select_stmt);
    }
    else if (auto insert_stmt = std::dynamic_pointer_cast<InsertStatement>(stmt)) {
        return generateInsertPlan(insert_stmt);
    }
    else if (auto update_stmt = std::dynamic_pointer_cast<UpdateStatement>(stmt)) {
        return generateUpdatePlan(update_stmt);
    }
    else if (auto delete_stmt = std::dynamic_pointer_cast<DeleteStatement>(stmt)) {
        return generateDeletePlan(delete_stmt);
    }
    else if (auto create_stmt = std::dynamic_pointer_cast<CreateTableStatement>(stmt)) {
        return generateCreateTablePlan(create_stmt);
    }

    return nullptr;
}

std::shared_ptr<PlanOperator> PlanGenerator::generateSelectPlan(std::shared_ptr<SelectStatement> stmt) {
    // 获取FROM子句中的表名
    std::string table_name;
    if (auto from_clause = std::dynamic_pointer_cast<FromClause>(stmt->from_clause)) {
        table_name = from_clause->table_name;
    }

    // 创建顺序扫描算子
    auto seq_scan = std::make_shared<SeqScanOperator>(table_name);

    // 如果有WHERE条件，添加过滤算子
    std::shared_ptr<PlanOperator> current = seq_scan;
    if (stmt->where_clause) {
        current = std::make_shared<FilterOperator>(current, stmt->where_clause);
    }

    // 添加投影算子
    current = std::make_shared<ProjectOperator>(current, stmt->select_list);

    return current;
}

std::shared_ptr<PlanOperator> PlanGenerator::generateInsertPlan(std::shared_ptr<InsertStatement> stmt) {
    return std::make_shared<InsertOperator>(stmt->table_name, stmt->values);
}

std::shared_ptr<PlanOperator> PlanGenerator::generateUpdatePlan(std::shared_ptr<UpdateStatement> stmt) {
    return std::make_shared<UpdateOperator>(stmt->table_name, stmt->assignments, stmt->where_clause);
}

std::shared_ptr<PlanOperator> PlanGenerator::generateDeletePlan(std::shared_ptr<DeleteStatement> stmt) {
    return std::make_shared<DeleteOperator>(stmt->table_name, stmt->where_clause);
}

std::shared_ptr<PlanOperator> PlanGenerator::generateCreateTablePlan(std::shared_ptr<CreateTableStatement> stmt) {
    return std::make_shared<CreateTableOperator>(stmt->table_name, stmt->columns);
}

// ================================
// 公开的查询执行接口
// ================================

std::vector<std::unordered_map<std::string, std::variant<std::string, int, double>>>
executeSelectQuery(std::shared_ptr<SelectStatement> select_stmt) {
    return global_executor.executeSelect(select_stmt);
}

} // namespace MiniDB
